// Copyright (C)  , Tencent. All rights reserved.

package blockworkpool

import (
	"context"
	"sync"
	"time"
)

/*
当前的SyncTaskGroup存在以下几个问题：
1，批次同步，如果任务执行时间不均匀会造成其他协程等待饿死，没有达到真正并发
2，同步任务装配在内存，许多任务可能长时间得不到同步，造成内存浪费

阻塞工作驰的改进：
生产消费模式，能够让最大协程满负荷工作，没有饿死情况发生
无需所有任务完全装配内存，降低内存消耗
*/

const (
	// 工作协程退出等待时间
	workerIdleTimeSeconds int = 2

	// 默认的缓存chan size
	bufferSize int = 10000
)

// 需要执行的任务
type Task func()

// 执行任务对应的执行函数，
// 增加该函数便于和coroutine池结合使用
func (t Task) Run() {
	t()
}

// 另一种接口执行函数
type RunTask interface {
	Run()
}

// coroutine pool 抽象
type Pool struct {
	sema chan int

	queue chan Task

	nextTask chan Task

	wg sync.WaitGroup

	mu sync.Mutex

	stopLoop bool
}

func runTask(t Task) {
	t()
}

// 终止接受新的任务
func (p *Pool) Stop() {
	p.mu.Lock()
	p.stopLoop = true
	p.mu.Unlock()

	close(p.nextTask)
	p.wg.Wait()
}

// 终止接受新的任务，并等待剩余任务执行完成
func (p *Pool) StopWait() {
	close(p.queue)
	p.wg.Wait()
}

func (p *Pool) slotsInUse() int {
	return len(p.sema)

}

func (p *Pool) worker() {
	var timer *time.Timer
	timer = time.NewTimer(1 * time.Second)
	defer func() {
		<-p.sema
		p.wg.Done()
	}()

	for {
		select {
		case t, ok := <-p.nextTask:
			if !ok {
				return
			}

			if t == nil {
				return
			}
			runTask(t)
			if !timer.Stop() {
				<-timer.C
			}
			timer.Reset(time.Duration(workerIdleTimeSeconds) * time.Second)
		case <-timer.C:
			return
		}
	}

}

func (p *Pool) tryNewWorker() {
	select {
	case p.sema <- 1:
		p.wg.Add(1)
		go p.worker()
	default:
	}
}

func (p *Pool) loop(ctx context.Context) {
	for {
		select {
		case t, ok := <-p.queue:
			if !ok {
				return //closed
			}
			if p.stopLoop {
				return
			}
			p.tryNewWorker()
			p.nextTask <- t
		case <-ctx.Done():
			return
		}
	}
}

// 提交任务
func (p *Pool) EnqueueTask(t Task) {
	select {
	case p.nextTask <- t:
		p.tryNewWorker()
	default:
		p.queue <- t
	}
}

// 创建新的coroutine pool
func NewPool(ctx context.Context, limit int, bfSize ...int) *Pool {
	if limit <= 0 {
		limit = 1
	}
	chanSize := bufferSize
	if len(bfSize) > 0 {
		chanSize = bfSize[0]
	}
	p := &Pool{
		queue:    make(chan Task, chanSize),
		sema:     make(chan int, limit),
		nextTask: make(chan Task, 1),
	}

	go p.loop(ctx)

	return p
}
